package main
/**
http://www.topgoer.com/%E6%95%B0%E6%8D%AE%E5%BA%93%E6%93%8D%E4%BD%9C/go%E6%93%8D%E4%BD%9Ckafka/kafka%E6%B7%B1%E5%B1%82%E4%BB%8B%E7%BB%8D.html
1.⽣产者从Kafka集群获取分区leader信息
2.⽣产者将消息发送给leader
3.leader将消息写入本地磁盘
4.follower从leader拉取消息数据
5.follower将消息写入本地磁盘后向leader发送ACK
6.leader收到所有的follower的ACK之后向生产者发送ACK

ACK应答机制
0代表producer往集群发送数据不需要等到集群的返回，不确保消息发送成功。安全性最低但是效 率最高。
1代表producer往集群发送数据只要leader应答就可以发送下一条，只确保leader发送成功。
all代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下一条，确保 leader发送成功和所有的副本都完成备份。安全性最⾼高，但是效率最低。
最后要注意的是，如果往不存在的topic写数据，kafka会⾃动创建topic，partition和replication的数量 默认配置都是1。
 */

/**
Partition：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞 吐量。同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的⽂件夹！
 */
import (
	"fmt"
	kafka "github.com/Shopify/sarama"
	"log"
)

func main() {
	//生产者
	config := kafka.NewConfig()
	config.Producer.RequiredAcks =kafka.WaitForAll  // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner=kafka.NewRandomPartitioner // 新选出一个partition
	config.Producer.Return.Successes=true  // 成功交付的消息将在success channel返回
	//构造消息体
	msg := &kafka.ProducerMessage{}
	msg.Topic= "web_log"
	msg.Value=kafka.StringEncoder("this is a test log")
	//连接kafka
	client,err := kafka.NewSyncProducer([]string{"127.0.0.1:9092"},config)
	if err!=nil {
		log.Fatal(err)
		return
	}
	defer client.Close()
	//发送消息
	pid, offset, err :=client.SendMessage(msg)
	if err != nil {
		fmt.Println("send msg failed, err:", err)
		return
	}
	fmt.Printf("pid:%v offset:%v\n", pid, offset)

}